package com.huang.hadoop;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.log4j.BasicConfigurator;

import java.io.IOException;
import java.util.Iterator;

public class FlowSum {
    public FlowSum() {
    }

    public static void main(String[] args) throws Exception {
        BasicConfigurator.configure();
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(FlowSum.class);
        job.setMapperClass(FlowSumMapper.class);
        job.setReducerClass(FlowSumReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        FileInputFormat.setInputPaths(job, new Path[]{new Path("input/MapReduce.txt")});
        FileOutputFormat.setOutputPath(job, new Path("output/result.txt"));
        boolean res = job.waitForCompletion(true);
        System.exit(res ? 0 : 1);
    }

    public static class FlowSumReducer extends Reducer<Text, FlowBean, Text,FlowBean> {
        FlowBean v = new FlowBean();

        public FlowSumReducer() {
        }

        protected void reduce(Text key, Iterable<FlowBean> values, Reducer<Text, FlowBean, Text, FlowBean>.Context context) throws IOException, InterruptedException {
            long upFlowCount = 0L;
            long downFlowCount = 0L;

            FlowBean bean;
            for(Iterator var8 = values.iterator(); var8.hasNext(); downFlowCount += bean.getDownFlow()) {
                bean = (FlowBean)var8.next();
                upFlowCount += bean.getUpFlow();
            }

            this.v.set(upFlowCount, downFlowCount);
            context.write(key, this.v);
        }
    }

    public static class FlowSumMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
        Text k = new Text();
        FlowBean v = new FlowBean();

        public FlowSumMapper() {
        }

        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = StringUtils.split(line, "\t");
            String phoneNum = fields[1];
            long upFlow = Long.parseLong(fields[fields.length - 3]);
            long downFlow = Long.parseLong(fields[fields.length - 2]);
            this.k.set(phoneNum);
            this.v.set(upFlow, downFlow);
            context.write(this.k, this.v);
        }
    }
}
